package org.duang.lock;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.duang.common.exceptios.ServiceException;

public class ZkSyncLock2 implements Lock, Watcher{

	private ZooKeeper zk;
	private String rootPath = "/locks";		// 根
	private String splitStr = "_lock_";		// 分割字符串
	private String lockName;				// 竞争资源的标志
	private String waitPrevLock;			// 等待前一个锁
	private String currentLock;				// 当前锁
	private CountDownLatch latch;			// 计数器
	private int sessionTimeout = 30000;
	private List<Exception> exceptions = new ArrayList<Exception>();
	
	public ZkSyncLock2(String config, String lockName){
		this.lockName = lockName;
		try{
			zk = new ZooKeeper(config, sessionTimeout, this);
			Stat stat = zk.exists(rootPath, false);
			if(null == stat){
				// 创建根节点
				zk.create(rootPath, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			}
		}catch (IOException e) {
			exceptions.add(e);
		} catch (KeeperException e) {
			exceptions.add(e);
		} catch (InterruptedException e) {
			exceptions.add(e);
		}
	}
	
	/**
	 * zookeeper节点的监视器
	 */
	@Override
	public void process(WatchedEvent event) {
		if(null != this.latch){
			this.latch.countDown();
		}
	}
	
	@Override
	public void lock() {
		if(exceptions.size() > 0){
			throw new ServiceException(exceptions.get(0));
		}
		try{
			if(this.tryLock()){
				System.out.println("Thread " + Thread.currentThread().getId() + " " +currentLock + " get lock true");
				return;
			} else {
				waitForLock(waitPrevLock, sessionTimeout);
			}
		} catch (KeeperException e) {
			exceptions.add(e);
		} catch (InterruptedException e) {
			exceptions.add(e);
		}
	}

	@Override
	public void lockInterruptibly() throws InterruptedException {
		this.lock();
	}

	@Override
	public Condition newCondition() {
		return null;
	}

	@Override
	public boolean tryLock() {
		
		if(lockName.contains(splitStr)){
			 throw new ServiceException("lockName can not contains \\u000B");
		}
		try {
			currentLock = zk.create(rootPath+"/"+lockName+splitStr, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			// 取出所有子节点
			List<String> subLockNames = zk.getChildren(rootPath, false);
			// 取出所有lockName的锁
			List<String> allLockNames = new ArrayList<String>();
			for(String subLockName : subLockNames){
				String tmpLockName = subLockName.split(splitStr)[0];
				if(tmpLockName.equals(lockName)){
					allLockNames.add(subLockName);
				}
			}
			Collections.sort(allLockNames);
			System.out.println(currentLock+ " == " + allLockNames.get(0));
			if(currentLock.equals(allLockNames.get(0))){
				// 如果是最小的节点, 则表示取得锁
				return true;
			}
			// 如果不是最小的节点，则找到比自己小1的节点
			String tmpLock  = currentLock.substring(currentLock.lastIndexOf("/")+1);
			waitPrevLock = allLockNames.get(Collections.binarySearch(allLockNames, tmpLock) - 1);
		} catch (KeeperException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		return false;
	}

	@Override
	public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
		if(this.tryLock()){
			return true;
		}
		try {
			return waitForLock(waitPrevLock, time);
		} catch (KeeperException e) {
			e.printStackTrace();
		}
		return false;
	}

	@Override
	public void unlock() {
		try{
			System.out.println("unlock: " + waitPrevLock);
			zk.delete(waitPrevLock, -1);
			waitPrevLock = null;
			zk.close();
		} catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
	}

	public boolean waitForLock(String prevLock, long waitTimeout) throws InterruptedException, KeeperException  {
		String tmpLock = rootPath+"/"+prevLock;
		Stat stat = zk.exists(tmpLock, true);
		if(null != stat){
			System.out.println("Thread " + Thread.currentThread().getId() + " waiting for " + tmpLock);
			this.latch = new CountDownLatch(1);
			this.latch.await(waitTimeout, TimeUnit.MILLISECONDS);
			this.latch = null;
		}
		return true;
	}
	
	 
}
